
#include "Interpreter.h"
#include <iostream>
#include <cassert>
#include <iterator>


int Interpreter::HandleDistinctAndOrderBy(const std::string& table, const bool distinct, 
		const bool orderby, const std::string& orderbyAttrib,
		const bool projection, const std::vector<std::string>& projectionAttribs,
		std::vector<Tuple>& resultTuples, std::string& resultName)
{
	// TODO: maybe the resultName isn't needed, as the schema of the resultTuples won't change

	Relation* relationPtr = mSchemaManager.getRelation(table);
	if(relationPtr == NULL)
		return -1;
	
	Schema schema = relationPtr->getSchema();
	
	bool intSort = false;
	if((orderby == true) && (schema.getFieldType(orderbyAttrib) == INT))
		intSort = true;


	if(relationPtr->getNumOfBlocks() <= mMem.getMemorySize())
	{
		// In memory operation won't need resultName... it'll be sent back empty

		std::cout << "In memory de-duping / sorting" << std::endl;
		
		relationPtr->getBlocks(0, 0, relationPtr->getNumOfBlocks());
		resultTuples = mMem.getTuples(0, relationPtr->getNumOfBlocks());

		if(distinct)
		{
			MultiSortFunctor* sorter;
			TupleEqualityFunctor* tupleEquals;

			std::vector<Tuple>::iterator it;
			if(projection == false)
			{
				std::vector<std::string> fieldNames = schema.getFieldNames();
				std::vector<enum FIELD_TYPE> fieldTypes = schema.getFieldTypes();

				sorter = new MultiSortFunctor(fieldNames, fieldTypes);

				tupleEquals = new TupleEqualityFunctor (fieldNames, fieldTypes);
			}
			else
			{
				std::vector<enum FIELD_TYPE> projectionFieldTypes(projectionAttribs.size());
				for(unsigned int fld = 0; fld < projectionAttribs.size(); ++fld)
					projectionFieldTypes[fld] = schema.getFieldType(projectionAttribs[fld]);

				sorter = new MultiSortFunctor(projectionAttribs, projectionFieldTypes);

				tupleEquals = new TupleEqualityFunctor(projectionAttribs, projectionFieldTypes);
			}

			std::sort(resultTuples.begin(), resultTuples.end(), *sorter);
			it = std::unique(resultTuples.begin(), resultTuples.end(), *tupleEquals);

			Tuple dummy = relationPtr->createTuple();
			resultTuples.resize(it - resultTuples.begin(), dummy);

			delete sorter;
			delete tupleEquals;

#if 0		/////// DEBUG //////////
			std::cout << "De-duped tuples are: " << std::endl;
			for(int i = 0; i < resultTuples.size(); ++i)
				std::cout << resultTuples[i] << std::endl;
			std::cout << std::endl;
#endif		///////////////////////
		}

		if(orderby)
		{

			SortFunctor sorter(orderbyAttrib, intSort);

			std::sort(resultTuples.begin(), resultTuples.end(), sorter);

#if 0		/////// DEBUG //////////
			std::cout << "OrderBy-ed tuples are: " << std::endl;
			for(int i = 0; i < resultTuples.size(); ++i)
				std::cout << resultTuples[i] << std::endl;
			std::cout << std::endl;
#endif		///////////////////////
		}
	}
	else
	{
		std::cout << "Two pass de-duping / sorting" << std::endl;

		// check if the two pass algorithm can do it ( B(R) <= M^2 )
		if(mSchemaManager.getRelation(table)->getNumOfBlocks() > 
			(mMem.getMemorySize() * mMem.getMemorySize()))
		{
			std::cout << "Table larger than M^2. Can't do two-pass... Exiting!" << std::endl;
			return -1;
		}
		//
		// Apparently the only sample query with both DISTINCT and ORDER BY will only be
		// tested for in-memory operations. That query also has projection in it. 
		// Otherwise, if projection was not involved, it'd be smarter to rearrange the 
		// fields such that the orderbyAttrib is the first field to be compared while sorting
		// for de-duping. This would automatically produce de-duped results which were sorted
		// for the orderbyAttrib, thus saving a good number of I/Os
		//

		// create new schema and relation for output, then create tuples
		resultName.assign(table);
	
		if(distinct)
		{
			resultName.append("_deDuped");

			// same schema as input
			if(mSchemaManager.createRelation(resultName, schema) == NULL)
				return -1;


			MultiSortFunctor* sorter;
			TupleEqualityFunctor* tupleEquals;

			if(projection == false)
			{
				std::vector<std::string> fieldNames = schema.getFieldNames();
				std::vector<enum FIELD_TYPE> fieldTypes = schema.getFieldTypes();

				sorter = new MultiSortFunctor(fieldNames, fieldTypes);
				tupleEquals = new TupleEqualityFunctor (fieldNames, fieldTypes);
			}
			else
			{
				std::vector<enum FIELD_TYPE> projectionFieldTypes(projectionAttribs.size());
				for(unsigned int fld = 0; fld < projectionAttribs.size(); ++fld)
					projectionFieldTypes[fld] = schema.getFieldType(projectionAttribs[fld]);

				sorter = new MultiSortFunctor(projectionAttribs, projectionFieldTypes);
				tupleEquals = new TupleEqualityFunctor(projectionAttribs, projectionFieldTypes);
			}

			std::string sortedSublists;
			SublistLimitsType sublistLimits;

			CreateSortedSublists(table, *sorter, sortedSublists, sublistLimits);

			RunDistinctPass2(sortedSublists, sublistLimits, *sorter, *tupleEquals, 
													resultName, resultTuples);

			delete sorter;
			delete tupleEquals;
		}

		if(orderby)
		{
			//
			// again it is clear that the tuples to perform orderby must be in a relation
			// on the disk.
			// Either distinct was never performed, and the table to be read is the table
			// input to this function
			// Or distinct was performed, in which case, we need to write its output to disk
			// coz it was assumed that the resultTuples of distinct went straight to the screen
			// without any more aggregation tasks performed on it
			//

			if(distinct)
			{
				AppendTuplesToRelation(resultName, resultTuples);
				resultTuples.clear();
			}

			std::string tableToSort(resultName); // either 'table' or 'table_sorted'
			resultName.append("_sorted"); // either 'table_sorted' or 'table_deDuped_sorted'

			// tableToSort must be deleted if(distinct == true) coz its an intermediate table

			// same schema as input
			if(mSchemaManager.createRelation(resultName, schema) == NULL)
				return -1;


			std::string sortedSublists;
			SublistLimitsType sublistLimits;
			PushClauseVecType pushClauses; // leave it empty, its harmless

			CreateSortedSublists(tableToSort, orderbyAttrib, sortedSublists, pushClauses, 
									sublistLimits);
			
			SortFunctor sorter(orderbyAttrib, intSort);
			RunOrderByPass2(sortedSublists, sublistLimits, sorter, resultName, resultTuples);

			if(distinct)
			{
				mSchemaManager.deleteRelation(tableToSort);
			}
		}

	}

	return 0;
}



int Interpreter::CreateSortedSublists(const std::string& table, const MultiSortFunctor& sorter,
			std::string& sortedTable, SublistLimitsType& sublistLimits)
{

	sortedTable.assign(table);
	sortedTable.append("_sortedSublists");

	Relation* relationPtr = mSchemaManager.getRelation(table);

	// create the first pass output relation with same schema as the input table
	if(mSchemaManager.createRelation(sortedTable, relationPtr->getSchema()) == NULL)
		return -1;

	Relation* sortedRelationPtr = mSchemaManager.getRelation(sortedTable);


	const int numBlocks = relationPtr->getNumOfBlocks();
	int blockCtr = 0;
	std::vector<Tuple> tuplesOfSublist;
	
	// technically there is scope for more pushing of selection here, but never mind
	while(blockCtr < numBlocks)
	{
		int blocksToCopy = std::min(numBlocks - blockCtr, mMem.getMemorySize());
		relationPtr->getBlocks(blockCtr, 0, blocksToCopy);
		blockCtr += blocksToCopy;

		tuplesOfSublist = mMem.getTuples(0, blocksToCopy);

		std::sort(tuplesOfSublist.begin(), tuplesOfSublist.end(), sorter);

		mMem.setTuples(0, tuplesOfSublist);

		// save the blockStart and blockEnd for this sublist, to be used by the Pass2()
		sublistLimits.push_back(std::make_pair(
								sortedRelationPtr->getNumOfBlocks(), 
								sortedRelationPtr->getNumOfBlocks() + blocksToCopy));

		// Since there are no holes and no pushed selection, the number of blocks to 
		// write after sorting will be equal to the number of blocks read
		sortedRelationPtr->setBlocks(sortedRelationPtr->getNumOfBlocks(), 0, blocksToCopy);
	}

#if 0	///// DEBUG /////

	std::cout << "The relation " << table << " contains:\n" << *relationPtr << std::endl;
	std::cout << "The sortedSublistRelation " << sortedTable << " contains:\n" 
		<< *sortedRelationPtr << std::endl;

#endif ///// DEBUG /////

	return 0;
}





int Interpreter::RunDistinctPass2(const std::string& sortedSublists,
	const SublistLimitsType& sublistLimits, const MultiSortFunctor& tupleLess,
	const TupleEqualityFunctor& tupleEquals, const std::string& resultTable,
	std::vector<Tuple>& resultTuples)
{

/*	 1) Find # of sublists
	 2) Allocate sublist vector
	 3) Initialize sublist by reading sublist block and filling queue
	 4) Pick smallest tuple of attrib attr from head of "all" sublists.
	 	If unable to pick a smallest tuple, all queues are empty, quit
	 5) Push one instance of the smallest tuple to output
		Remove those heads from sublists that have this smallest tuple
		If a queue becomes empty, refill it by reading from the disk if possible
	 	Then go to step (4)
*/

	//////////////////////////
	// INITIALIZE
	//////////////////////////

	Relation* relationPtr = mSchemaManager.getRelation(sortedSublists);;
	std::vector<Sublist> sublistVec(sublistLimits.size());

	for(unsigned int sub = 0; sub < sublistVec.size(); ++sub)
	{
		Sublist& sublist = sublistVec[sub];

		sublist.blockIdx = sublistLimits[sub].first;
		sublist.blockEnd = sublistLimits[sub].second;

		sublist.memBlockIdx = sub; // 0,1,2...

		relationPtr->getBlock(sublist.blockIdx, sublist.memBlockIdx);
		std::vector<Tuple> tempTuples = mMem.getTuples(sublist.memBlockIdx, 1);
		for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
		{
			sublist.tuples.push(tempTuples[tpl]);
		}
	}

	//////////////////////////
	// RUN ALGORITHM
	//////////////////////////

	Tuple resultTuple = mSchemaManager.getRelation(resultTable)->createTuple();
	std::vector<enum FIELD_TYPE> resultFieldTypes = resultTuple.getSchema().getFieldTypes();

	Schema schema = relationPtr->getSchema();

	const int MAX_INT = std::numeric_limits<int>::max();
	const std::string MAX_STR("zzzzzzzzzz");
	
	// create max tuple for default comparison
	// both input schema and result schema are same
	Tuple MAX_TUPLE = mSchemaManager.getRelation(resultTable)->createTuple();
	for(int i = 0; i < MAX_TUPLE.getNumOfFields(); ++i)
	{
		if(resultFieldTypes[i] == INT)
			MAX_TUPLE.setField(i, MAX_INT);
		else
			MAX_TUPLE.setField(i, MAX_STR);
	}

	while(1)
	{
		resultTuple = MAX_TUPLE;

		// find the smallest tuple from the heads of sublists
		for(unsigned int sub = 0; sub < sublistVec.size(); ++sub)
		{
			Sublist& sublist = sublistVec[sub];

			if(sublist.tuples.empty() == false)
			{
				if(tupleLess(sublist.tuples.front(), resultTuple) == true)
					resultTuple = sublist.tuples.front();
			}
		}

		// check if all queues are empty, so we can exit this massive while(1)
		if(tupleEquals(resultTuple, MAX_TUPLE))
			break;


		// add this smallest tuple to resultTuples
		resultTuples.push_back(resultTuple);

		// delete from heads of those queues which have the smallest tuple
		for(unsigned int sub = 0; sub < sublistVec.size(); ++sub)
		{
			Sublist& sublist = sublistVec[sub];

			while((sublist.tuples.empty() == false) && 
					(tupleEquals(sublist.tuples.front(), resultTuple) == true))
			{
				//
				// the smallet tuples could be across the 
				// block boundary, hence if popping makes a block empty, it
				// needs to be read and queue refilled right here, if its possible
				// to do so
				//
				sublist.tuples.pop();

				if(sublist.tuples.empty())
				{
					++sublist.blockIdx;
					if(sublist.blockIdx < sublist.blockEnd)
					{
						relationPtr->getBlock(sublist.blockIdx, sublist.memBlockIdx);
						std::vector<Tuple> tempTuples = 
									mMem.getTuples(sublist.memBlockIdx, 1);
						for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
						{
							sublist.tuples.push(tempTuples[tpl]);
						}
					}
				}
			}
		}
	} // end while

	// the sorted sublists aren't needed any more
	mSchemaManager.deleteRelation(sortedSublists);

	return 0;
}


int Interpreter::RunOrderByPass2(const std::string& sortedSublists,
	const SublistLimitsType& sublistLimits, const SortFunctor& tupleLess,
	const std::string& resultTable, std::vector<Tuple>& resultTuples)
{

/*	 1) Find # of sublists
	 2) Allocate sublist vector
	 3) Initialize sublist by reading sublist block and filling queue
	 4) Pick smallest tuple of attrib attr from head of "all" sublists.
	 	If unable to pick a smallest tuple, all queues are empty, quit
	 5) Push smallest tuple to output
		Remove the from that sublist that has this smallest tuple
		If a queue becomes empty, refill it by reading from the disk if possible
	 	Then go to step (4)
*/

	//////////////////////////
	// INITIALIZE
	//////////////////////////

	Relation* relationPtr = mSchemaManager.getRelation(sortedSublists);;
	std::vector<Sublist> sublistVec(sublistLimits.size());

	for(unsigned int sub = 0; sub < sublistVec.size(); ++sub)
	{
		Sublist& sublist = sublistVec[sub];

		sublist.blockIdx = sublistLimits[sub].first;
		sublist.blockEnd = sublistLimits[sub].second;

		sublist.memBlockIdx = sub; // 0,1,2...

		relationPtr->getBlock(sublist.blockIdx, sublist.memBlockIdx);
		std::vector<Tuple> tempTuples = mMem.getTuples(sublist.memBlockIdx, 1);
		for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
		{
			sublist.tuples.push(tempTuples[tpl]);
		}
	}

	//////////////////////////
	// RUN ALGORITHM
	//////////////////////////

	Tuple resultTuple = mSchemaManager.getRelation(resultTable)->createTuple();
	std::vector<enum FIELD_TYPE> resultFieldTypes = resultTuple.getSchema().getFieldTypes();

	Schema schema = relationPtr->getSchema();

	const int MAX_INT = std::numeric_limits<int>::max();
	const std::string MAX_STR("zzzzzzzzzz");
	
	// create max tuple for default comparison
	// both input schema and result schema are same
	Tuple MAX_TUPLE = mSchemaManager.getRelation(resultTable)->createTuple();
	for(int i = 0; i < MAX_TUPLE.getNumOfFields(); ++i)
	{
		if(resultFieldTypes[i] == INT)
			MAX_TUPLE.setField(i, MAX_INT);
		else
			MAX_TUPLE.setField(i, MAX_STR);
	}

	while(1)
	{
		resultTuple = MAX_TUPLE;

		// find the smallest tuple from the heads of sublists and note its sublist #
		int smallestSublistNum = -1;
		for(unsigned int sub = 0; sub < sublistVec.size(); ++sub)
		{
			Sublist& sublist = sublistVec[sub];

			if(sublist.tuples.empty() == false)
			{
				if(tupleLess(sublist.tuples.front(), resultTuple) == true)
				{
					resultTuple = sublist.tuples.front();
					smallestSublistNum = sub;
				}
			}
		}

		// check if all queues are empty, so we can exit this massive while(1)
		if(smallestSublistNum == -1)
			break;

		// add this smallest tuple to resultTuples
		resultTuples.push_back(resultTuple);

		Sublist& sublist = sublistVec[smallestSublistNum];
		
		// pop the sublist that has this smallest tuple
		sublist.tuples.pop();

		// if the sublist became empty with this pop, reload it from the disk if possible
		if(sublist.tuples.empty())
		{
			++sublist.blockIdx;
			if(sublist.blockIdx < sublist.blockEnd)
			{
				relationPtr->getBlock(sublist.blockIdx, sublist.memBlockIdx);
				std::vector<Tuple> tempTuples = 
							mMem.getTuples(sublist.memBlockIdx, 1);
				for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
				{
					sublist.tuples.push(tempTuples[tpl]);
				}
			}
		}
	} // end while

	// the sorted sublists aren't needed any more
	mSchemaManager.deleteRelation(sortedSublists);




	return 0;
}






